package com.xiaoran.client.run;

import com.xiaoran.common.data.RpcResponse;
import io.netty.channel.EventLoopGroup;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Promise;
import org.springframework.stereotype.Component;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Rpc异步任务处理
 */
@Component
public class RpcRequestPool {

    private final ConcurrentHashMap<String, Promise<RpcResponse>> requestPool = new ConcurrentHashMap<>();

    public void submitRequest(String requestId, EventExecutor executor) {
        requestPool.put(requestId, new DefaultPromise<>(executor));
    }

    public RpcResponse fetchResponse(String requestId) throws Exception {
        Promise<RpcResponse> promise = requestPool.get(requestId);
        if (promise == null) {
            return null;
        }
        RpcResponse rpcResponse = promise.get(10, TimeUnit.SECONDS);
        requestPool.remove(requestId);
        RpcRequestManager.destroyChannelHolder(requestId);

        return rpcResponse;
    }

    public void notifyRequest(String requestId, RpcResponse rpcResponse) {
        Promise<RpcResponse> promise = requestPool.get(requestId);
        if (promise != null) {
            promise.setSuccess(rpcResponse);
        }
    }
}
